package com.gmall.realtime.dim.function;

import com.alibaba.fastjson.JSONObject;
import com.gmall.realtime.common.bean.TableProcessDim;
import com.gmall.realtime.common.constant.Constant;
import com.gmall.realtime.common.util.HbaseUtil;
import com.gmall.realtime.common.util.RedisUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.hadoop.hbase.client.Connection;
import redis.clients.jedis.Jedis;

import java.io.IOException;

@Slf4j
public class HBaseSinkFunction extends RichSinkFunction<Tuple2<JSONObject, TableProcessDim>> {

    private Connection conn;
    private Jedis jedis;

    @Override
    public void open(Configuration parameters) throws Exception {
        // 获取HBase连接
        conn = HbaseUtil.getHBaseConnection();
//        jedis = RedisUtil.getJedis();
    }


    @Override
    public void close() throws Exception {
        // 关闭HBase连接
        HbaseUtil.closeHBaseConn(conn);
//        RedisUtil.closeJedis(jedis);
    }


    @Override
    public void invoke(Tuple2<JSONObject, TableProcessDim> dataWithConfig,
                       Context context) throws Exception {
        // insert update delete bootstrap-insert
        // 获取数据
        JSONObject data = dataWithConfig.f0;
        // 获取操作类型
        String opType = data.getString("op_type");

        if ("delete".equals(opType)) {
            // 删除维度信息
            delDim(dataWithConfig);
        }else{
            // insert update 和 bootstrap-insert 的时候, 写入维度数据
            // 写入维度数据
            putDim(dataWithConfig);
        }

        TableProcessDim tableProcessDim = dataWithConfig.f1;
        // 如果是维度是 update 或 delete 则删除缓存中的维度数据
        if ("delete".equals(opType) || "update".equals(opType)) {
            String key = RedisUtil.getKey(
                    tableProcessDim.getSinkTable(),
                    data.getString(tableProcessDim.getSinkRowKey()));
//            jedis.del(key);
        }
    }


    /**
     * 向HBase表中写入维度数据
     *
     * @param dataWithConfig 包含维度数据和表配置信息的元组
     *                       - f0: JSON格式的维度数据
     *                       - f1: 表处理配置信息
     * @throws IOException 当HBase操作发生错误时抛出
     */
    private void putDim(Tuple2<JSONObject, TableProcessDim> dataWithConfig) throws IOException {
        // 从元组中提取数据部分和表配置信息
        JSONObject data = dataWithConfig.f0;
        TableProcessDim tableProcessDim = dataWithConfig.f1;

        // 获取行键：从数据中根据表配置指定的行键字段获取值
        String rowKeyField = tableProcessDim.getSinkRowKey();
        String rowKey = data.getString(rowKeyField);
        if (rowKey == null) {
            // 记录更详细的错误日志，包含表名、行键字段和原始数据
            log.error("RowKey is null for table: {}, rowKeyField: {}, data: {}",
                    tableProcessDim.getSinkTable(), rowKeyField, data);
            return;
        }

        // 记录写入HBase的日志信息
        log.info("向 HBase 表 {} 写入数据，rowKey: {}", tableProcessDim.getSinkTable(), rowKey);
        // 移除操作类型字段，避免将其写入HBase
        data.remove("op_type");

        // 调用HBase工具类写入数据
        HbaseUtil.putRow(conn,
                Constant.HBASE_NAMESPACE,
                tableProcessDim.getSinkTable(),
                rowKey,
                tableProcessDim.getSinkFamily(),
                data);
    }


    private void delDim(Tuple2<JSONObject, TableProcessDim> dataWithConfig) throws IOException {
        // 获取Tuple2的第一个元素，即JSONObject类型的数据
        JSONObject data = dataWithConfig.f0;
        // 获取Tuple2的第二个元素，即TableProcessDim类型的表处理维度信息
        TableProcessDim tableProcessDim = dataWithConfig.f1;

        // 从JSONObject中获取与表处理维度信息中的sinkRowKey对应的值，作为行键
        String rowKey = data.getString(tableProcessDim.getSinkRowKey());

        // 调用HbaseUtil的delRow方法，删除HBase中指定行键的行
        // 参数依次为：HBase连接对象、HBase命名空间、表名、行键
        HbaseUtil.delRow(conn,
                Constant.HBASE_NAMESPACE,
                tableProcessDim.getSinkTable(),
                rowKey);
    }

}